import json
from kafka import KafkaProducer
from datetime import datetime, timedelta
import random
import string

# ===================== Kafka配置 =====================
KAFKA_BROKERS = "node101:9092,node102:9092,node103:9092"  # 替换为实际Kafka地址
KAFKA_TOPIC = "ev_monitor_topic"  # 目标Topic

# 初始化Kafka Producer
producer = KafkaProducer(
    bootstrap_servers=KAFKA_BROKERS,
    value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode('utf-8'),  # 支持中文
    retries=3,  # 失败重试
    batch_size=16384,  # 批量大小（16KB）
    linger_ms=10  # 等待时间，优化批量发送
)


# ===================== 随机数据生成函数 =====================
def generate_vin():
    """生成符合GB16735的17位VIN码（示例逻辑，可根据实际规则调整）"""
    world_manufacturer = random.choice(["LFW", "LGW", "LJV"])  # 中国厂商代码示例
    vehicle_descriptor = ''.join(random.choices(string.ascii_uppercase + string.digits, k=8))
    model_year = random.choice(["2", "3", "4", "5"])  # 年份代码（简化）
    plant_code = random.choice(string.ascii_uppercase)
    check_digit = random.randint(0, 9)
    production_sequence = f"{random.randint(100000, 999999):06d}"
    return f"{world_manufacturer}{vehicle_descriptor}{model_year}{plant_code}{check_digit}{production_sequence}"


def generate_mileage_diagnosis():
    odometer = random.randint(1000, 300000)  # 仪表里程 1000-300000 km
    actual = int(odometer * random.uniform(0.5, 1.2))  # 实测里程允许±50%偏差（模拟真实场景）
    deviation = round((actual - odometer) / odometer * 100, 2) if odometer != 0 else 0.0

    # 先定义vehicle_age_hours变量
    vehicle_age_hours = random.randint(500, 50000)  # 车龄 500-50000小时（约5.7年-5.7年）
    usage_hours = random.randint(100, vehicle_age_hours)  # 用车时长 ≤ 车龄
    reasonable_usage_hours = random.randint(usage_hours, vehicle_age_hours)  # 合理时长 ≥ 用车时长

    return {
        "tampering_risk": random.choice([True, False]),  # 50%概率有调表风险
        "vehicle_age_hours": vehicle_age_hours,
        "odometer_mileage": odometer,
        "actual_mileage": actual,
        "mileage_deviation": deviation,
        "usage_hours": usage_hours,
        "reasonable_usage_hours": reasonable_usage_hours,
        "mileage_consistency": random.choice(["一致", "轻微偏差", "明显偏差"]),  # 随机一致性评价
        "tampering_risk_desc": "该车辆存在调表变更里程风险" if random.choice([True, False]) else "该车辆不存在调表变更里程风险"
    }


def generate_fault_diagnosis():
    """故障与路况诊断（随机故障+随机路况）"""
    fault_status = random.choice(["无故障", "电池故障", "电机故障", "传感器故障", "电控系统故障"])
    road_condition = random.choice(["城市道路", "高速公路", "乡村道路", "山路", "拥堵", "通畅", "施工路段", "暂无数据"])
    return {
        "fault_status": fault_status,
        "road_condition": road_condition
    }


def generate_region_analysis(num_regions=3):
    """行驶区域分析（随机区域+动态数量）"""
    regions = ["京", "津", "冀", "鲁", "沪", "苏", "浙", "粤", "川", "渝"]
    return [
        {
            "region_code": random.choice(regions),
            "distance_km": random.randint(1, 5000),  # 行驶距离 1-5000 km
            "duration_hours": random.randint(1, 200)  # 行驶时长 1-200小时
        } for _ in range(random.randint(1, num_regions))  # 每车1-3个区域（动态数量）
    ]


def generate_time_distribution(num_periods=7):
    """时段分布（随机时段+动态数量）"""
    periods = ["凌晨", "早晨", "上午", "中午", "下午", "傍晚", "深夜"]
    selected_periods = random.sample(periods, k=num_periods)  # 随机选择7个时段（可修改数量）
    return [
        {
            "time_period": p,
            "distance_km": random.randint(1, 3000),  # 时段里程 1-3000 km
            "duration_hours": random.randint(1, 24)  # 时段时长 1-24小时
        } for p in selected_periods
    ]


def generate_driving_behavior():
    """驾驶行为分析（正态分布评分模拟）"""
    trip_count = random.randint(10, 10000)  # 历史行程 10-10000次
    driver_score = int(random.normalvariate(mu=70, sigma=15))  # 评分正态分布（均值70，标准差15）
    driver_score = max(0, min(100, driver_score))  # 确保评分在0-100之间
    comparison = random.choice([
        f"高于同车型{random.randint(30, 90)}%用户",
        f"低于同车型{random.randint(10, 50)}%用户",
        "与同车型平均水平相当"
    ])
    return {
        "trip_count": trip_count,
        "driver_score": driver_score,
        "score_comparison": comparison,
        "weekly_data": [
            {
                "week": f"第{week}周",
                "distance_km": random.randint(100, 2000),
                "score": int(random.normalvariate(mu=driver_score, sigma=10))
            } for week in random.sample(range(1, 53), k=4)  # 随机4周数据
        ]
    }


def generate_maintenance_records():
    """保养记录（随机时间+里程）"""
    now = datetime.now()
    maintenance_dates = [now - timedelta(days=random.randint(30, 1000)) for _ in range(2)]  # 最近2次保养日期
    return {
        "maintenance_records": [
            {
                "maintenance_date": md.strftime('%Y-%m-%d'),
                "mileage_km": random.randint(1000, 200000),  # 保养时里程
                "engine_hours": random.randint(0, 10000),  # 发动机时长
                "is_recommended": random.choice([True, False])  # 50%概率为建议保养
            } for md in maintenance_dates
        ],
        "next_maintenance": {
            "date": (now + timedelta(days=random.randint(30, 365))).strftime('%Y-%m-%d'),  # 下次保养日期
            "distance_remaining": random.randint(500, 5000),  # 剩余里程
            "days_remaining": random.randint(30, 180),  # 剩余天数
            "engine_hours_remaining": random.randint(100, 2000)  # 剩余发动机时长
        }
    }


def generate_weather_distribution(num_weathers=5):
    """天气分布（随机天气类型+动态数量）"""
    weathers = ["晴", "阴", "雨", "雪", "雾", "多云", "小雨", "中雨", "大雨", "小雪", "中雪", "其他"]
    return [
        {
            "weather_type": w,
            "distance_km": random.randint(1, 2000),  # 天气下行驶里程
            "duration_hours": random.randint(1, 100)  # 天气下行驶时长
        } for w in random.sample(weathers, k=num_weathers)  # 随机5种天气（可修改数量）
    ]


# ===================== 生成并发送数据 =====================
def main():
    total = 200
    for i in range(total):
        data = {
            "vin": generate_vin(),
            "analysis_title": "联网智能分析",
            "report_time": datetime.now().isoformat(),
            "mileage_diagnosis": generate_mileage_diagnosis(),
            "fault_diagnosis": generate_fault_diagnosis(),
            "region_analysis": generate_region_analysis(),
            "time_distribution": generate_time_distribution(),
            "driving_behavior": generate_driving_behavior(),
            "maintenance_records": generate_maintenance_records(),
            "weather_distribution": generate_weather_distribution(),
            "create_time": datetime.now().isoformat(),
            "update_time": datetime.now().isoformat()
        }

        # 发送至Kafka
        try:
            future = producer.send(KAFKA_TOPIC, data)
            future.get(timeout=10)  # 同步等待发送确认
            print(f"已发送第{i + 1}/{total}条数据，VIN: {data['vin'][:8]}...")
        except Exception as e:
            print(f"发送失败: {str(e)}")

    producer.flush()  # 确保所有消息发送完成
    producer.close()  # 关闭Producer
    print(f"\n成功发送{total}条完全随机的新能源汽车监控数据至Kafka Topic: {KAFKA_TOPIC}")


if __name__ == "__main__":
    main()